/*
package com.seaboxdata.threadHandler;

import com.rabbitmq.client.*;
import com.seaboxdata.kafka.BaseStringKafkaProducer;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.io.IOException;

*/
/**
 * @author
 * @create 2020-09-10 16:25
 **//*


@Component
@Slf4j
public class RealTimeBusCashRabbitMqHandler implements DisposableBean, Runnable {

    private Thread thread;

    private boolean flag;

    private Address[] addresses;

    private ConnectionFactory factory;

    @Value("${realTimeBusCash.rabbitMq.ip}")
    private String rabbitMqIp;

    @Value("${realTimeBusCash.rabbitMq.port}")
    private String port;

    @Value("${realTimeBusCash.rabbitMq.username}")
    private String username;

    @Value("${realTimeBusCash.rabbitMq.password}")
    private String password;

    @Value("${realTimeBusCash.rabbitMq.queueName}")
    private String queueName;

    @Autowired
    private BaseStringKafkaProducer baseStringKafkaProducer;

    @Value("${realTimeBusCash.kafka.realTimeBusCashTopic}")
    private String topic;

    private Connection connection = null;

    private Channel channel = null;

    @Autowired
    public void initThread() {
        addresses = new Address[]{
                new Address(rabbitMqIp, Integer.valueOf(port))
        };
        this.factory = new ConnectionFactory();
        factory.setHost(rabbitMqIp);
        factory.setPort(Integer.valueOf(port));
        factory.setUsername(username);
        factory.setPassword(password);
        this.flag = true;
        try {
            this.connection = factory.newConnection(addresses);
            this.channel = connection.createChannel();
        } catch (Exception e) {
            e.printStackTrace();
            log.error("connect rabbit mq error");
        }
        this.thread = new Thread(this);
        thread.setName("RealTimeBusCashRabbitMqHandler");
        //this.thread.start();
    }

    @Override
    public void run() {
        try {
            Consumer consumer = new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    try {
                        baseStringKafkaProducer.sendMessageToKafka(new String(body), topic);
                    } catch (Exception e) {
                        e.printStackTrace();
                        log.error("write kafka error");
                    }
                }
            };
            channel.basicConsume(queueName, consumer);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    @Override
    public void destroy() throws Exception {
        this.flag = false;
    }
}
*/
